package com.example.ranking.listener;

import com.example.ranking.util.RankingGlobalConfig;
import com.example.ranking.listener.PetInfoCanalClient.PetStatusChangeMsg;
import com.example.ranking.domain.po.PetPopularRankingPO;
//import com.example.ranking.mapper.PetPopularRankingMapper;
import com.example.ranking.config.MqExchangeQueueConfig;
import com.example.ranking.mapper.RankingMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;

import java.util.concurrent.TimeUnit;

import static com.example.ranking.config.MqExchangeQueueConfig.PET_STATUS_QUEUE;

/**
 * MQ消费者：处理宠物状态变更消息，增量更新人气榜统计
 */
@Component
@Slf4j
@RequiredArgsConstructor
public class PetStatusChangeConsumer {

    private final RankingMapper popularRankingMapper;
    private final RedissonClient redissonClient;
    private final MqExchangeQueueConfig rabbitMqConfig;

    // 分布式锁前缀（避免多实例重复更新同一品种）
    private static final String LOCK_KEY_PREFIX = "ranking:breed:lock:";
    // 锁超时时间（5秒，避免死锁）
    private static final long LOCK_TIMEOUT = 5;

    /**
     * 监听宠物状态变更队列
     */
    @RabbitListener(queues = PET_STATUS_QUEUE) // 配置文件中对应RabbitMqConfig.PET_STATUS_QUEUE
    @Transactional(rollbackFor = Exception.class)
    public void consumePetStatusChangeMsg(PetStatusChangeMsg msg) throws InterruptedException {
        // 1. 校验初始化状态（未初始化完成则拒绝消费）
        if (!RankingGlobalConfig.isMqConsumeEnabled()) {
            log.warn("全量初始化未完成，拒绝消费消息：{}", msg);
            Thread.sleep(3000);
            throw new RuntimeException("全量初始化未完成，消息暂不消费"); // 触发MQ重试机制
        }

        // 2. 校验消息合法性
        Assert.hasText(msg.getPetBreed(), "宠物品种不能为空");
        Assert.hasText(msg.getChangeType(), "变更类型不能为空");

        // 3. 获取分布式锁（按品种加锁，避免多实例同时更新同一品种）
        String lockKey = LOCK_KEY_PREFIX + msg.getPetBreed();
        var lock = redissonClient.getLock(lockKey);
        try {
            // 尝试加锁（最多等待3秒，加锁后5秒自动释放）
            boolean locked = lock.tryLock(3, LOCK_TIMEOUT, TimeUnit.SECONDS);
            if (!locked) {
                log.error("获取分布式锁失败，消息消费失败：{}", msg);
                throw new RuntimeException("获取锁失败，消息重试"); // 触发MQ重试
            }

            // 4. 查询该品种的当前统计记录
            PetPopularRankingPO breedPO = popularRankingMapper.selectByBreed(msg.getPetBreed());
            int countChange = "ADD".equals(msg.getChangeType()) ? 1 : -1;

            // 5. 更新统计数量（存在则更新，不存在则新增）
            if (breedPO != null) {
                // 防止数量为负（如状态从3变为非3，但统计记录已为0）
                int newCount = Math.max(0, breedPO.getCurrentCount() + countChange);
                breedPO.setCurrentCount(newCount);
                popularRankingMapper.updateById(breedPO);
                log.info("更新品种统计：{}，原数量={}，变更={}，新数量={}",
                        msg.getPetBreed(), breedPO.getCurrentCount() - countChange, countChange, newCount);
            } else {
                // 仅当变更类型为ADD时新增（REMOVE时无记录则无需处理）
                if ("ADD".equals(msg.getChangeType())) {
                    PetPopularRankingPO newPO = new PetPopularRankingPO();
                    newPO.setPetBreed(msg.getPetBreed());
                    newPO.setCurrentCount(1);
                    popularRankingMapper.insert(newPO);
                    log.info("新增品种统计：{}，初始数量=1", msg.getPetBreed());
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("消费消息中断：{}", msg, e);
            throw new RuntimeException("消息消费中断");
        } finally {
            // 6. 释放锁（确保锁一定释放）
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}
